#include <iterator>
#include <unistd.h>
#include "MarkupSTL.h"
#include "global.h"
#include "cmd.h"
#include "Game.h"
#include "RedisClient.h"
#include "Logic.h"
#include "json.h"
#include "globalExtern.h"
#include "Host.h"
#include <string>
#include <algorithm>
#include "log.h"
#include "util.h"
#include "base64.h"
#include "Md5.h"
#include "bsl_string.h"
#include <sstream>
#include <map>
#include <time.h>
#include <cstdlib>
#include "public_define.h"
#include "cache.h"
#include "DbPart.h"
#include "apiordermgr.h"
#include "minutelisten.h"
#include "NodeServer.h"
#include "3400_api_order.pb.h"
#include "LogHandler.h"

#define  MOD_UID (100)
char* CGameServer::s_szexchange = "exceptionflow";
char* CGameServer::s_szqueue = "exceptionflowhandle";


CGameServer::CGameServer() : m_pthread(NULL), m_mqchannel(1), m_nodecount(0), m_nodeindex(-1), m_exitflag(false), m_initflag(false), m_cachecheckflag(false)
{
}

CGameServer::~CGameServer()
{
	if (m_pthread)
	{
		m_exitflag = true;
		m_pthread->join();
		delete m_pthread;
		m_pthread = NULL;
	}

	if (m_mqconn)
		releasemq();
}

int CGameServer::OnBYTimeOut(int Timerid)
{
	switch (Timerid)
	{
	case TIMERID_CACHE_CHECK:
	{
		procTimerCacheCheck();
		break;
	}
	case TimerID::TIMER_MINITE:
		MinuteListen::Instance()->OnTick();
		StartTimer(TimerID::TIMER_MINITE, 60, false);
		break;
	case TimerID::TIMER_API_ORDER_CREATE:
		ProcApiOrderExceptionCreatePush();
		break;
	case TimerID::TIMER_API_ORDER_FINISH:
		ProcApiOrderExceptionFinishPush();
		break;
	default:
		break;
	}
	return 0;
}

bool CGameServer::InitGame()
{
	log_info("game init");
	char conf_path[128] = "./config.xml";
	if (!ReadCommonConf(conf_path))
	{
		log_boot("InitGame ReadCommonConf failed. path:[%s]", conf_path);
		return false;
	}

	//StartTimer(TimerID::TIMER_MINITE, 60, false);
	return initmq();
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPwd = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPwd.empty() || 0 == m_mqPort)
		return false;

	log_debug("ReadCommonConf succ!!");

	return true;
}


int CGameServer::ProcessPacket(SocketHandler * pHandler, NETInputPacket* inputPkg)
{
	do
	{
		int cmd = inputPkg->GetCmdType();
		log_debug("cmd=0x%x,  svid=%d, ", cmd, pHandler->getId());

		if (0 == LogHandler::Instance()->ProcessPacket(m_pHost, pHandler, inputPkg))
		{
			return 0;
		}

		switch (cmd)
		{
			case api_order::API_ORDER_CMD_ORDER_EXCEPTION_CREATE_PUSH:
				log_debug("ProcApiOrderExceptionCreatePush init");
				//ProcApiOrderExceptionCreatePush();
				break;
			case api_order::API_ORDER_CMD_ORDER_EXCEPTION_FINISH_PUSH:
				log_debug("ProcApiOrderExceptionFinishPush init");
				//ProcApiOrderExceptionFinishPush();
				break;
			default:
				log_error("cmd=0x%x,  svid=%d, err no proc", cmd, pHandler->getId());
				break;
		}
		return 1;
	}while(0);
	return -1;
}

void CGameServer::ProcApiOrderExceptionCreatePush()
{
	int cnt = 100;
	std::vector<api_order::ApiOrder> api_order_group;
	do
	{
		api_order::ApiOrder api_order;
		if (!Cache::Instance()->RpopApiOrderCreate(api_order))
		{
			break;
		}
		log_debug("api_order:%s", api_order.ShortDebugString().c_str());
		api_order_group.push_back(api_order);
	} while (cnt--);
	// 多条批量插入
	if (!api_order_group.empty())
	{
		ApiOrderMgr::Instance()->BatchCreateExceptionOrder(api_order_group);
	}
	// 有积压则继续处理
	if (cnt == 0)
	{
		StartTimerMillisecond(TimerID::TIMER_API_ORDER_CREATE, 1, false);
	}
}
void CGameServer::ProcApiOrderExceptionFinishPush()
{
	int cnt = 100;
	do
	{
		api_order::ApiOrder api_order;
		if (!Cache::Instance()->RpopApiOrderFinish(api_order))
		{
			break;
		}
		log_debug("api_order:%s", api_order.ShortDebugString().c_str());
		ApiOrderMgr::Instance()->FinishExceptionOrder(api_order);
	} while (cnt--);
	// 有积压则继续处理
	if (cnt == 0)
	{
		StartTimerMillisecond(TimerID::TIMER_API_ORDER_FINISH, 1, false);
	}
}

bool CGameServer::initmq(void)
{
	m_mqconn = amqp_new_connection();
	if (!m_mqconn)
	{
		log_error("create mq conn err");
		return false;
	}
	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn);
	if (!socket)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("create mq TCP socket err");
		return false;
	}
	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
	if (status)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq socket err");
		return false;
	}
	amqp_rpc_reply_t x = amqp_login(m_mqconn, m_mqVhost.c_str(), 0, 131072, 30, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
	if (AMQP_RESPONSE_NORMAL != x.reply_type)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
		return false;
	}
	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn, m_mqchannel);
	if (!t)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq channel fail");
		return false;
	}

	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
	{
		releasemq();
		log_error("mq channel amqp_get_rpc_reply fail");
		return false;
	}

	return true;
}

bool CGameServer::mqbind(void)
{
	if (!(m_nodecount > 0))
		return false;
	if (!(m_nodeindex < m_nodecount))
		return false;

	//bool bindflag = false;
	char szqueue[50] = { 0 };
	::snprintf(szqueue, sizeof(szqueue) - 1, "%s%u", s_szqueue, TGlobal::_svid);
	for (int i = 0; i < MOD_UID; ++i)
	{
		if (m_nodeindex == i % m_nodecount)
		{
			amqp_queue_declare_ok_t* r = amqp_queue_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), false, true, false, false, amqp_empty_table);
			if (!r)
			{
				releasemq();
				log_error("mq amqp_queue_declare fail");
				return false;
			}

			amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x2.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_declare fail");
				return false;
			}

			amqp_queue_bind(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), amqp_cstring_bytes(CGameServer::s_szexchange), amqp_cstring_bytes(std::to_string(i).c_str()), amqp_empty_table);
			amqp_rpc_reply_t x = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_bind fail,exchange:%s,queue:%s,routing:%d", CGameServer::s_szexchange, szqueue, i);
				return false;
			}
			//bindflag = true;
			log_debug("amqp_queue_bind ok,exchange:%s,queue:%s,routing:%d", CGameServer::s_szexchange, szqueue, i);

			amqp_basic_consume(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), amqp_empty_bytes, false, true, false, amqp_empty_table);
			amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x1.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_bind fail,exchange:%s,queue:%s", CGameServer::s_szexchange, szqueue);
				return false;
			}
		}
	}

	return true;
}
void CGameServer::releasemq(void)
{
	if (m_mqconn)
	{
		amqp_channel_close(m_mqconn, m_mqchannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
	}
	return;
}

void CGameServer::run(void)
{
	struct timeval timeout;
	timeout.tv_sec = 2;
	timeout.tv_usec = 0;
	amqp_frame_t frame;
	while (!m_exitflag)
	{
		if (!m_mqconn)
		{
			if (!initmq())
			{
				log_error("mq reconnect fail,sleep 1");
				::sleep(1);
				continue;
			}
			if (!mqbind())
			{
				log_error("mq rebind fail,sleep 1");
				::sleep(1);
				continue;
			}
		}

		amqp_envelope_t envelope;
		amqp_maybe_release_buffers(m_mqconn);
		amqp_rpc_reply_t ret = amqp_consume_message(m_mqconn, &envelope, &timeout, 0);
		if (AMQP_RESPONSE_NORMAL == ret.reply_type)
		{
			procMessage((const char*)envelope.message.body.bytes, envelope.message.body.len);
			amqp_destroy_envelope(&envelope);
		}
		else
		{
			if (AMQP_RESPONSE_LIBRARY_EXCEPTION == ret.reply_type)
			{
				if (AMQP_STATUS_TIMEOUT == ret.library_error)
				{
				}
				else if (AMQP_STATUS_HEARTBEAT_TIMEOUT == ret.library_error)
				{
					log_debug("heart beat");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_SOCKET_ERROR == ret.library_error)
				{
					log_error("sock err");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_CONNECTION_CLOSED == ret.library_error)
				{
					log_error("connection closed");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_UNEXPECTED_STATE == ret.library_error)
				{
					if (AMQP_STATUS_OK != amqp_simple_wait_frame(m_mqconn, &frame))
					{
						log_error("amqp_simple_wait_frame fail");
						releasemq();
						if (!initmq())
						{
							log_error("mq reconnect fail");
							continue;
						}
						if (!mqbind())
						{
							log_error("mq rebind fail");
							continue;
						}
						continue;
					}

					if (AMQP_FRAME_METHOD == frame.frame_type)
					{
						switch (frame.payload.method.id)
						{
						case AMQP_BASIC_ACK_METHOD:
							/* if we've turned publisher confirms on, and we've published a
							 * message here is a message being confirmed.
							 */
							break;
						case AMQP_BASIC_RETURN_METHOD:
							/* if a published message couldn't be routed and the mandatory
							 * flag was set this is what would be returned. The message then
							 * needs to be read.
							 */
						{
							amqp_message_t message;
							ret = amqp_read_message(m_mqconn, frame.channel, &message, 0);
							if (AMQP_RESPONSE_NORMAL != ret.reply_type)
							{
								break;
							}

							amqp_destroy_message(&message);
						}

						break;

						case AMQP_CHANNEL_CLOSE_METHOD:
							/* a channel.close method happens when a channel exception occurs,
							 * this can happen by publishing to an exchange that doesn't exist
							 * for example.
							 *
							 * In this case you would need to open another channel redeclare
							 * any queues that were declared auto-delete, and restart any
							 * consumers that were attached to the previous channel.
							 */
						{
							log_error("channel close method");
							releasemq();
							if (!initmq())
							{
								log_error("mq reconnect fail");
								continue;
							}
							if (!mqbind())
							{
								log_error("mq rebind fail");
								continue;
							}
						}
						break;

						case AMQP_CONNECTION_CLOSE_METHOD:
							/* a connection.close method happens when a connection exception
							 * occurs, this can happen by trying to use a channel that isn't
							 * open for example.
							 *
							 * In this case the whole connection must be restarted.
							 */
						{
							log_error("connection close method");
							releasemq();
							if (!initmq())
							{
								log_error("mq reconnect fail");
								continue;
							}
							if (!mqbind())
							{
								log_error("mq rebind fail");
								continue;
							}
						}
						break;
						default:
							log_error("An unexpected method was received %u", frame.payload.method.id);
							break;
						}
					}
				}
				else
				{
					log_error("unexpected err,%d", ret.library_error);
				}
			}
			else if (AMQP_RESPONSE_SERVER_EXCEPTION == ret.reply_type)
			{
				if (AMQP_CHANNEL_CLOSE_METHOD == ret.reply.id || AMQP_CONNECTION_CLOSE_METHOD == ret.reply.id)
				{
					log_error("reply.id:%x", ret.reply.id);
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
			}
		}
		if (m_cachecheckflag)
		{
			chachCheck();
			m_cachecheckflag = false;
		}
	}

	return;
}

int CGameServer::procMessage(const char* data, const unsigned int len)
{
	if (!data || 0 == len)
		return 1;

	api_order::ApiOrder msg;
	if (!msg.ParseFromArray(data, len))
	{
		log_error("parse pb fail,len:%u", len);
		return 2;
	}

	time_t tt = ::time(NULL);
	std::string tablename;

	unsigned int uid = msg.uid();
	if (0 == uid)
	{
		log_error("uid=0");
		return 3;
	}
	_u64_ time = msg.start_time();
	if (0 == time)
		time = tt;

	tableName(uid, g_uid_mod, time, tablename);
	std::map<std::string, _u64_>::const_iterator itor1 = m_tables.find(tablename);
	if (m_tables.end() == itor1)
	{
		if (!CDbPart::Instance()->createApiOrderExceptionTable(tablename))
		{
			log_error("add table fail,%s", tablename.c_str());
			return 4;
		}
		m_tables.insert(std::make_pair(tablename, tt));  
	}
	if (tablename.empty())
	{
		log_error("wrong tablename,uid:%u", uid);
		return 5;
	}

	log_debug("apiorder exception,uid,%d,act,%d", uid, msg.act());

	if (api_order::API_ORDER_ACT_CREATE == msg.act())
	{
		CDbPart::Instance()->addApiOrderException(tablename, msg);
	}
	else if (api_order::API_ORDER_ACT_DELETE == msg.act())
	{
		CDbPart::Instance()->finishApiOrderException(tablename, msg, 1);

		tableName(uid, g_uid_mod, msg.start_time() - 86400, tablename);
		std::map<std::string, _u64_>::const_iterator itor1 = m_tables.find(tablename);
		if (m_tables.end() == itor1)
		{
			if (!CDbPart::Instance()->createApiOrderExceptionTable(tablename))
			{
				log_error("add table fail,%s", tablename.c_str());
				return 4;
			}
			m_tables.insert(std::make_pair(tablename, msg.start_time() - 86400));
		}

		if (tablename.empty())
		{
			log_error("wrong tablename,uid:%u", uid);
			return 5;
		}

		CDbPart::Instance()->finishApiOrderException(tablename, msg, 1);
	}

	return 0;
}

int CGameServer::ProcEventPublishService(int ev_id, SpiderEvent* pEvent)
{
	if (!(pEvent && pEvent->evParam.ptr))
		return -1;

	if (m_pthread)
		return 0;

	ServiceConf* pserviceConf = (ServiceConf*)pEvent->evParam.ptr;
	if (SERVER_TYPE_API_ORDER_MANAGER == pserviceConf->server_type)
	{
		ServiceTypeList::const_iterator itor = m_pHost->m_pNode->m_serviceTypeList.find(SERVER_TYPE_API_ORDER_MANAGER);
		if (m_pHost->m_pNode->m_serviceTypeList.end() != itor && itor->second.size() > 0)
		{
			int index = 0;
			for (ServiceVect::const_iterator itor1 = itor->second.begin(); itor->second.end() != itor1; ++itor1, ++index)
			{
				if (TGlobal::_svid == itor1->svid)
				{
					break;
				}
			}


			if (index < itor->second.size())
			{
				m_nodecount = itor->second.size();
				m_nodeindex = index;

				log_debug("srv id,%d,index,%d,service num,%d", TGlobal::_svid, index, itor->second.size());
				if (m_nodecount > 0)
				{
					if (mqbind())
					{
						m_pthread = new std::thread([this](void) {
							run();
							});
					}
					else
					{
						exit(-1);
						return false;
					}
				}
				else
				{
					log_error("wrong srv num,srv id,%d,index,%d,service num,%d", TGlobal::_svid, index, itor->second.size());
				}
			}
			else
			{
				log_error("service find no srv id,srv id,%d", TGlobal::_svid);
			}
			log_debug("service pulish ready, srv id,%d", TGlobal::_svid);
		}
		else
		{
			log_error("service pulish not ready");
		}
	}
	return 0;
}

int CGameServer::ProcEventGameInit(int ev_id, SpiderEvent* pEvent)
{
	if (!m_initflag)
	{
		if (0 != m_pHost->InitMtSend())
		{
			log_error("InitMtSend fail");
			exit(-1);
			return -1;
		}

		StartTimer(TIMERID_CACHE_CHECK, g_cache_check_dur, true);
		m_initflag = true;
	}

	return 0;
}

int CGameServer::procTimerCacheCheck(void)
{
	m_cachecheckflag = true;

	return 0;
}

void CGameServer::chachCheck(void)
{
	time_t tt = ::time(NULL);
	for (std::map<std::string, _u64_>::iterator itor = m_tables.begin(); m_tables.end() != itor; )
	{
		if (tt > g_cache_dur + itor->second)
		{
			m_tables.erase(itor++);
		}
		else
		{
			++itor;
		}
	}
}